package com.yunhe.realtime.websocket;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.realtime.service.data.RealtimeService;
import com.yunhe.common.websocket.WebsocketRealtimeHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 量测实时数据WebSocket处理类
 * @author liuronglei
 */
public class RealtimeHandler extends WebsocketRealtimeHandler {
    private static final ObjectMapper mapper = new ObjectMapper();
    // Json类型转换定义
    private TypeReference measurementTR = new TypeReference<List<Measurement>>(){};
    @Autowired
    private RealtimeService realtimeService;

    @Override
    public void beforeSend(WebSocketSession session, Object content) {

    }

    @Override
    public void afterEstablished(WebSocketSession session) {

    }

    @Override
    public void afterClosed(WebSocketSession session, CloseStatus status) {

    }

    @Override
    public void handleMessage(WebSocketSession session, TextMessage message, List<String> points) {
        Map<String, Double> result = realtimeService.findMeasureRealtimeByStr(points);
        if (result != null) {
            sendMessage(result, session, points);
        }
    }

    @KafkaListener(topics = KafkaConsts.Topic.Realtime, groupId = KafkaConsts.Group.RandomGroupId)
    public void realtimeListener(ConsumerRecord<String, String> record) throws Exception {
        String value = record.value();
        if (value != null) {
            List<Measurement> list = mapper.readValue(value, measurementTR);
            if (list != null && list.size() > 0) {
                List<String> keys = new ArrayList<>();
                Map<String, Double> values = new HashMap<>();
                for (Measurement measurement : list) {
                    Long pointNumber = measurement.getPointNumber();
                    if (pointNumber != null) {
                        String pid = String.valueOf(pointNumber);
                        keys.add(pid);
                        values.put(pid, measurement.getVal());
                    }
                }
                sendMessage(values, null, keys);
            }
        }
    }
}